package com.bingxu.flink.chapter02.transformation;

import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;

/**
 * @author :BingXu
 * @description :TODO
 * @date :2021/8/9 11:34
 * @modifier :
 */

public class Trans_Connect {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(3);
        DataStreamSource<Integer> intDS = env.fromElements(1, 2, 3, 5, 6, 2, 21, 8, 91, 45);
        DataStreamSource<String> strDS = env.fromElements("hello", "world", "hello", "flink", "spark");

        ConnectedStreams<Integer, String> connectedStreams = intDS.connect(strDS);
        connectedStreams.getFirstInput().print("Integer--->>>>");
        connectedStreams.getSecondInput().print("String--->>>>");

        SingleOutputStreamOperator<String> mapCoDS = connectedStreams.map(new CoMapFunction<Integer, String, String>() {
            @Override
            public String map1(Integer integer) throws Exception {
                return integer.toString();
            }

            @Override
            public String map2(String s) throws Exception {
                return s;
            }
        });
        mapCoDS.print();

        env.execute();

    }
}
